package com.hp.bon.sgw.threads;

import com.hp.bon.sgw.Constants;
import com.hp.bon.sgw.core.CallParam;
import com.hp.bon.sgw.domain.BusinessRecord;
import com.hp.bon.sgw.service.SGWDataStoreService;
import com.hp.bon.sgw.start.SpringContext;
import com.hp.bon.sgw.util.Comutils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.concurrent.ConcurrentLinkedQueue;

public class BusinessRecordCacheManager extends TaskThread
{
    public static final Logger LOG = LoggerFactory.getLogger(BusinessRecordCacheManager.class);

    private static BusinessRecordCacheManager _instance;
  private ArrayList<CallParam> readList;
  private ArrayList<CallParam> restList;
  private static int BUFFER_MAX_NUM = 10240;
  private static long MAX_LAZY_TIME = 10000L;

  public LinkedList<CallParam> lazyObjectBuffer = new LinkedList<CallParam>();
  public ConcurrentLinkedQueue<ArrayList<CallParam>> logObjectBufferQueue = new ConcurrentLinkedQueue<ArrayList<CallParam>>();
  private String datapath;
  private boolean saveDB = false;

  public static BusinessRecordCacheManager getInstance()
  {
    if (_instance == null) {
      _instance = new BusinessRecordCacheManager();
    }
    return _instance;
  }

  public BusinessRecordCacheManager()
  {
    setName("Task-BusinessRecordCache");
    this.sleepTime = 512L;

    for (int i = 0; i < 3; i++) {
      ArrayList<CallParam> list = new ArrayList<CallParam>(BUFFER_MAX_NUM);
      this.logObjectBufferQueue.add(list);
    }
    this.readList = ((ArrayList<CallParam>)this.logObjectBufferQueue.poll());
    this.restList = ((ArrayList<CallParam>)this.logObjectBufferQueue.poll());

    this.datapath = System.getProperty("business_data_path");
    this.saveDB = ("db".equalsIgnoreCase(System.getProperty("business_data_save")));
    Constants.LOG.info("business data save to DB: " + this.saveDB + " ,business_data_path is " + this.datapath);
  }

  public void add(CallParam o)
  {
    ArrayList<CallParam> writeList = (ArrayList<CallParam>)this.logObjectBufferQueue.peek();
    boolean isBusy = false;
    if (writeList.size() < BUFFER_MAX_NUM) {
      synchronized (writeList) {
        writeList.add(o);
      }
      o.setInQueueTime(System.currentTimeMillis());
    } else {
      isBusy = true;
    }

    if (isBusy)
      flush2File(o);
  }

  protected void process()
  {
    List<BusinessRecord> tempList = new LinkedList<BusinessRecord>();

    long minWaitTimeStamp = this.runTime - MAX_LAZY_TIME;
    ListIterator<CallParam> it = this.lazyObjectBuffer.listIterator();
    BusinessRecord str;
    while (it.hasNext()) {
      CallParam callParam = (CallParam)it.next();
      if ((callParam.getSendRespTime() != 0L) || (callParam.getInQueueTime() < minWaitTimeStamp)) {
        str = SGWDataStoreService.convertCallParamToBusinessRecord(callParam);
        if (str != null) {
          tempList.add(str);
        }
        it.remove();
      }

    }

    if (this.readList != null && this.readList.size() > 0) {
//        LOG.info("process readList:"+  new Gson().toJson(readList));
      for (CallParam callParam : this.readList) {
          if(callParam == null){
              LOG.error("process error callParam is null");
            continue;
          }
        if ((callParam.getResponse() != null) && (callParam.getSendRespTime() == 0L)) {
          this.lazyObjectBuffer.add(callParam);
        }
        else {
           str = SGWDataStoreService.convertCallParamToBusinessRecord(callParam);
          if (str != null)
            tempList.add(str);
        }
      }
      this.readList.clear();
    }

    this.logObjectBufferQueue.offer(this.readList);
    this.readList = this.restList;
    this.restList = ((ArrayList<CallParam>)this.logObjectBufferQueue.poll());

    if (tempList.size() > 0) {
      flush2BD(tempList);
      if (Constants.LOG.isDebugEnabled()) {
        StringBuilder sb = new StringBuilder();
        sb.append(getName()).append(" insert into DB records num =");
        sb.append(tempList.size()).append(",time(ms)=").append(System.currentTimeMillis() - this.runTime);
        Constants.LOG.debug(sb.toString());
      }
    }
  }

  protected void processBeforeExit() {
    ArrayList<CallParam> firstList = (ArrayList<CallParam>)this.logObjectBufferQueue.poll();
    List<BusinessRecord> tempList = new LinkedList<BusinessRecord>();
    for (CallParam callParam : firstList) {
      BusinessRecord str = SGWDataStoreService.convertCallParamToBusinessRecord(callParam);
      if (str != null) {
        tempList.add(str);
      }
    }
    for (CallParam callParam : this.lazyObjectBuffer) {
      BusinessRecord str = SGWDataStoreService.convertCallParamToBusinessRecord(callParam);
      if (str != null) {
        tempList.add(str);
      }
    }
    flush2BD(tempList);
  }

  private void flush2BD(List<BusinessRecord> tempList) {
    if (this.saveDB)
      SpringContext.getInstance().getOuterDBDataService().bulkSaveToDB(tempList);
    else
      saveDataToDisk(tempList);
  }

  private void flush2File(CallParam callParam)
  {
    BusinessRecord br = SGWDataStoreService.convertCallParamToBusinessRecord(callParam);
    if (br != null)
    {
      Constants.LOG.info("BusinessRecord=" + br.toString());
    }
  }

  private void saveDataToDisk(List<BusinessRecord> batchData) {
    long begin = System.currentTimeMillis();
    FileWriter outStream = null;
    try {
      outStream = getoutStream();
      for (BusinessRecord data : batchData) {
        if (outStream != null) {
          try {
            outStream.write(data.toString());
            outStream.write("\r\n");
          } catch (Exception e) {
            Constants.LOG.warn("write data file err:" + e.toString());
          }
        }
        data = null;
      }
    }
    catch (Exception e) {
      Constants.LOG.warn("cuaght err:" + e.toString());

      if (outStream != null)
        try {
          outStream.flush();
          outStream.close();
        }
        catch (Exception localException1)
        {
        }
    }
    finally
    {
      if (outStream != null) {
        try {
          outStream.flush();
          outStream.close();
        }
        catch (Exception localException2)
        {
        }
      }
    }

    long used = System.currentTimeMillis() - begin;

    if (used > 5000L)
      Constants.LOG.warn("write file too slow ,record num=" + batchData.size() + ",used time ms=" + used);
  }

  private FileWriter getoutStream()
  {
    String curfileName = Comutils.datafileFormat() + ".sgwdata";
    try {
      File file = new File(this.datapath);
      if (!file.exists()) {
        boolean succ = file.mkdirs();
        Constants.LOG.info(datapath + " not exist, do mkdir :" + succ);
      }
      return new FileWriter(new File(file, curfileName), true);
    }
    catch (IOException e) {
      Constants.LOG.warn("create out file err:" + e);
    }
    return null;
  }
}